<?php
/**
 *
 * @author fzq
 * @date 2016-06-27
 */

namespace libraries;
use enums\QueueEnums;
use libraries\CommUtils;

if( !defined( 'APP_ROOT' ) ){ echo 'Direct Access Deny!'; return; }

class RQueue implements \Phalcon\Di\InjectionAwareInterface, \libraries\IMsgQueue
{
	//array('mtype' => BIZ_SMS,
	//		'priority' =>  PRIORITY_HIGH,
	//		'cls' => 'Log',
	//		'method' => 'memLoginLog',
	//		'msg' => 'xxxxxxxxx' );
	
// 	private $_config;
	private $_di;
	private $_queue;
	
	/* (non-PHPdoc)
	 * @see \Phalcon\Di\InjectionAwareInterface::setDI()
	 */
	public function setDI(\Phalcon\DiInterface $dependencyInjector) 
	{
		// TODO Auto-generated method stub
		$this->_di = $dependencyInjector;
	}

	/* (non-PHPdoc)
	 * @see \Phalcon\Di\InjectionAwareInterface::getDI()
	 */
	public function getDI() 
	{
		// TODO Auto-generated method stub
		return $this->_di;
	}
	
	/**
	 * 
	 * @param mixed $msg
	 * @param string $biz
	 * @param string $cls
	 * @param string $method
	 * @param string $id
	 * @param int $priority
	 * @param int $msgt  QueueEnums::MSGTYPE_LONG ...
	 * 
	 * return false for error
	 */
	public function getMsgProto( $msg, 
								$strBiz, 
								$strCls, 
								$strMethod = 'defaultOp', 
								$id = false, 
								$priority = QueueEnums::PRIORITY_MEDIUM,
								$msgt = QueueEnums::MSGTYPE_STRING
								  )
	{
		if( !is_string( $strBiz ) || !is_string( $strCls ) || !is_string( $strMethod ) )
		{
			return false;
		}
		
		switch( $msgt )
		{
			case QueueEnums::MSGTYPE_STRING:
				$arrParams[ 'msg' ] = $msg;
				break;
			case QueueEnums::MSGTYPE_ARRAY:
			case QueueEnums::MSGTYPE_OBJEDCT:
				$arrParams[ 'msg' ] = serialize( $msg );
				break;
			case QueueEnums::MSGTYPE_LONG:
				$arrParams[ 'msg' ] = '' . $msg;
				break;
			default:
				return false;
		}
		
		$arrParams[ 'biz' ] = $strBiz;
		$arrParams[ 'cls' ] = $strCls;
		$arrParams[ 'method' ] = $strMethod;
		$arrParams[ 'id' ] = $id ? $id : CommUtils::getUNID( $strBiz . $strCls. $strMethod );
		$arrParams[ 'msgt' ] = $msgt;
		$arrParams[ 'priority' ] = $priority;
		
		return $arrParams;
		
	}
	
	//array('biz' => BIZ_SMS,
	//		'priority' =>  PRIORITY_HIGH,
	//		'id'	=> CommUtils::UNID(),
	//		'cls' => 'Log',
	//		'method' => 'memLoginLog',
	//		'msgt'	=> 'string,array,object',
	//		'msg' => 'xxxxxxxxx' );
	/**
	 * send message to message queue
	 * 
	 * @param array $arrParam
	 * return id for true false for error
	 */
	public function push( array $arrParam )
	{
		if( !is_array( $arrParam ) || 
			!isset( $arrParam[ 'biz' ] ) || 
		    !isset( $arrParam[ 'priority' ] ) || 
			!isset( $arrParam[ 'cls' ] ) || 
			!isset( $arrParam[ 'method' ] ) || 
			!isset( $arrParam[ 'msg' ] ) ||
			!isset( $arrParam[ 'msgt' ]) || 
			!isset( $arrParam[ 'id' ]) )
		{
			return false;
		}
		
		$lListLen = $this->_queue->lpush( 'q_' . $arrParam[ 'biz' ], serialize( $arrParam ));
		
		if( $lListLen )
		{
			return $arrParam[ 'id' ];
		}
		
		return false;
	}
	

	/**
	 * 从message queue中取数据
	 * @param string $strBiz
	 * @param string $strPriority
	 * @param string $bPush
	 * 
	 * 当使用信号处理时不能使用阻塞方式连接否则会出错
	 * return false for error string for success
	 */
	public function pop( $strBiz, $iPriority = QueueEnums::PRIORITY_MEDIUM, $bBlock = false, $timeout = 10, $bBackup = false )
	{
		if( !is_string( $strBiz ) )
		{
			return false;
		}
		
		if( $bBlock )
		{
			if( $timeout < 0 )
			{
				$timeout = 10;
			}
			
			if( $bBackup )
			{
				$arrMsg = $this->_queue->brpoplpush( 'q_' . $strBiz, 'q_b_' . $strBiz, $timeout );
			}
			else
			{
				$arrMsg = $this->_queue->brpop( 'q_' . $strBiz, $timeout );
			}

			if( $arrMsg )
			{
				return $arrMsg[1];
			}
			
			return false;
		}
		else
		{
			if( $bBackup )
			{
				return $this->_queue->rpoplpush( 'q_' . $strBiz, 'q_b_' . $strBiz );
			}
			else
			{
				return $this->_queue->rpop( 'q_' . $strBiz );
			}
		}
		
	}
	
	/**
	 * delete from backup queue
	 * @param array $arrParams
	 * 
	 * return false for error
	 */
	public function delFromBackupQueue( $arrParams )
	{
		return $this->_queue->lrem( 'q_b_' . $arrParams[ 'biz' ], serialize( $arrParams ) );
	}

	public function setServer( $queue )
	{
		if( isset( $this->_queue ))
		{
			unset( $this->_queue );
		}
		
		$this->_queue = $queue;
	}
	
// 	public function reconnect()
// 	{
// 		$redis = new \Redis();
		
// 		$config = $this->_di->get( 'config' );
		
		
// 		try 
// 		{
// 			$redis->pconnect( $config->rqueue->host, $config->rqueue->port /*, 'password'*/ );
// 		}
// 		catch( \Exception $e )
// 		{
// 			echo $e->getMessage();
			
// 			return false;	
// 		}
		
// 		$redis->select( $config->rqueue->index );
// 		$this->_queue->setServer( $redis );
		
// 		return true;
// 	}
	
	public function getServer()
	{
		return $this->_queue;
	}
	
// 	public function getConfig()
// 	{
// 		return $this->_config;
// 	}

// 	public function setConfig( $config )
// 	{
// 		$this->_config = $config;
// 	}
}
